package GEMonitor;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.File;
import java.io.FileOutputStream;
import java.time.LocalDate;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.Future;

public class GEKafkaProducer {

    private static KafkaProducer<String, String> producer;

    private static final String TOPIC = "geChannel";

    private static final String WAV_TOPIC = "geWavChannel";

    private static final String METER_TOPIC = "geMeterChannel";

    private static final String BOOTSTRAP_SERVERS_CONFIG = "kafka:9092";

    private static IdWorker worker = new IdWorker(1,1,1);


    static {
        // 构建连接配置
        Properties properties = new Properties();

        // 若要配多个服务器，用逗号隔开
        // 注：服务器要开放端口，若云服务器还要在server.properties配置内网IP和外网IP
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS_CONFIG);
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "my-producer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        properties.put(ProducerConfig.ACKS_CONFIG,"0");
//        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,1);
        // 构造Client无非是：new 或 工厂模式
        producer = new KafkaProducer<String, String>(properties);
    }

    private GEKafkaProducer(){

    }

    private static int[] instFloatStringData(byte[] response) {
        int[] result = new int[response.length * 2];
        for (int i = 0; i < response[8] / 2; i++) {
            result[i] = ((response[i * 2 + 9] << 8) + response[i * 2 + 10]);
        }
        return result;
    }

    public static void send(String systemId, String msgType, byte[] content){
        try {
            long now = System.currentTimeMillis();

            String contentStr = null;
            String topic = "";
            switch (msgType) {
                case "Z3":
                    topic = WAV_TOPIC;
                    break;
                case "HZ":
                    topic = METER_TOPIC;
                    break;
                default:
                    topic = TOPIC;
            }

//        String topic = "Z3".equals(msgType)?WAV_TOPIC:TOPIC;
            if (topic.equals(WAV_TOPIC)) {
                String s = new String(content);
                String[] ss = s.split(":");
                contentStr = ss[0] + ":" + ss[1];
                if (ss.length > 2) {
                    String fileName = worker.nextId() + ".data";
                    contentStr = contentStr + ":" + fileName;
                    byte[] fileContent = Arrays.copyOfRange(content, content.length - 88200, content.length);
                    String filePath = String.format("/mnt/filearchive/wav-data/%s/%s/%s", LocalDate.now().toString(), systemId, fileName);
                    File f = new File(filePath);
                    if (!f.getParentFile().exists()) {
                        f.getParentFile().mkdirs();
                    }
                    try (FileOutputStream file = new FileOutputStream(filePath)) {
                        file.write(fileContent);
                    } catch (Exception e) {
                        MyLogger.Error("文件生成失败", e);
                    }
                }

            } else {
                contentStr = new String(content);
            }
            contentStr = now + ":" + contentStr;
            int partition = Math.abs((systemId.hashCode() & 0x7FFFFFFF))%3+1;
            Future<RecordMetadata> f = producer.send(new ProducerRecord<String, String>(topic,partition, systemId, msgType + ":" + contentStr));
            RecordMetadata metadata = null;
            try {
                metadata = f.get();
            } catch (Exception e) {
                MyLogger.Error("发送失败",e);
            }
//            MyLogger.Info(systemId + "发送消息完成,partition:" + metadata.partition() + ", offset:" + metadata.offset() + ", content:" + contentStr);
        } catch(Throwable t){
            MyLogger.Error(systemId+"发送消息失败", t);
        }
    }
}


